package com.settlement.system.service.impl;

import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.model.entity.KafkaConsumeRecord;
import com.settlement.system.mapper.KafkaConsumeRecordMapper;
import com.settlement.system.service.KafkaConsumeRecordService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.RequiredArgsConstructor;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.springframework.util.Assert;
import com.settlement.system.common.util.StringUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.settlement.system.model.query.KafkaConsumeRecordQuery;
import com.settlement.system.model.form.KafkaConsumeRecordForm;
import org.springframework.beans.BeanUtils;
import com.settlement.system.model.vo.KafkaConsumeRecordImportVo;

/**
 * <p>
 * kafka消费记录 服务实现类
 * </p>
 *
 * @author huangrensen
 * @since 2024-08-24
 */
@Service
@RequiredArgsConstructor
public class KafkaConsumeRecordServiceImpl extends ServiceImpl<KafkaConsumeRecordMapper, KafkaConsumeRecord> implements KafkaConsumeRecordService {

    private final Logger logger = LoggerFactory.getLogger(KafkaConsumeRecordServiceImpl.class);

    private final RedissonClient redissonClient;

    @Value("${system.config.expire}")
    private Integer expire;
    @Override
    public Page<KafkaConsumeRecord> getPage(KafkaConsumeRecordQuery queryParams) {
        // 查询参数
        int pageNum = queryParams.getPageNum();
        int pageSize = queryParams.getPageSize();
        // 查询数据
        Page<KafkaConsumeRecord> page = new Page<>(pageNum, pageSize);
        return this.baseMapper.getPage(page, queryParams);
    }

     @Override
     public boolean saveKafkaConsumeRecord(KafkaConsumeRecordForm kafkaConsumeRecordForm) {
         KafkaConsumeRecord kafkaConsumeRecord = new KafkaConsumeRecord();
         BeanUtils.copyProperties(kafkaConsumeRecordForm, kafkaConsumeRecord);
         kafkaConsumeRecord.setCreateTime(null);
         return this.saveOrUpdate(kafkaConsumeRecord);
     }

    /**
     * 此方法用于生成kafka消费记录，返回true则代表没有消费过，如果返回false代表已经消费过，无需重复消费
     * @param dataId 需要保存的数据
     * @return 是否可以消费
     */
    @Override
    public boolean checkCanConsume(String dataId) {
        // 获取锁对象
        RLock lock = redissonClient.getLock(SystemConstants.CONSUMER_LOCK_PREFIX + dataId);
        if (Objects.isNull(expire)) {
            expire = 30;
        }
        try{
            boolean lockResult = lock.tryLock(20, expire, TimeUnit.SECONDS); // 获取锁失败，需要记录失败日志，方便后面重试
            if(lockResult){
                logger.info("获取锁成功：{}", dataId);
                List<Integer> yearList = new ArrayList<>();
                int year = LocalDate.now().getYear();
                int month = LocalDate.now().getMonthValue();
                // 只查最近两年的即可
                yearList.add(year);
                // 这里进一步优化，当当前月份时小于等于5月份的时候才查上一年的记录，提高速度
                if(month <= 5){
                    logger.info("由于当前月份小于等于5月份，所以上一年的消费记录也一并查询判断...");
                    yearList.add(year - 1);
                }
                logger.info("查询的kafka消费记录表有：【%s】".formatted(yearList));
                String tableName;
                KafkaConsumeRecord query;
                List<KafkaConsumeRecord> list;
                for (Integer yearInt : yearList) {
                    tableName = "kafka_consume_record_%s".formatted(yearInt);
                    // 检查是否消费过
                    if(StringUtils.hasText(this.baseMapper.existTable(tableName))){
                        query = new KafkaConsumeRecord();
                        query.setTableName(tableName);
                        query.setDataId(dataId);
                        list = this.baseMapper.getList(query);
                        if(Objects.nonNull(list) && !list.isEmpty()){
                            // 存在记录，直接返回false
                            return false;
                        }
                    }
                }
                // 返回可以消费的标记
                return true;
            }
        }catch (Exception e){
            logger.error("获取锁失败，原因：{}", e.getMessage());
        }finally {
            if (Objects.nonNull(lock) && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
        return true;
    }

    @Override
    public boolean saveData(String dataId) {
        KafkaConsumeRecord kafkaConsumeRecord = new KafkaConsumeRecord();
        // 保存到当年的分表中
        String tableName = "kafka_consume_record_%s".formatted(LocalDate.now().getYear());
        kafkaConsumeRecord.setTableName(tableName);
        kafkaConsumeRecord.setDataId(dataId);
        // 处理dataId超长
        if(kafkaConsumeRecord.getDataId().length() > 100){
            kafkaConsumeRecord.setDataId(kafkaConsumeRecord.getDataId().substring(0, 100));
        }
        // 检查是否存在表
        if(!StringUtils.hasText(this.baseMapper.existTable(tableName))){
            this.baseMapper.createTable(tableName);
        }
        return this.baseMapper.insertData(kafkaConsumeRecord) > 0;
    }

    @Override
     public KafkaConsumeRecordForm getKafkaConsumeRecordFormData(Long id) {
         KafkaConsumeRecord kafkaConsumeRecord = this.getById(id);
         KafkaConsumeRecordForm kafkaConsumeRecordForm = new KafkaConsumeRecordForm();
         BeanUtils.copyProperties(kafkaConsumeRecord, kafkaConsumeRecordForm);
         return kafkaConsumeRecordForm;
     }

     @Override
     public boolean deleteKafkaConsumeRecord(String ids) {
        Assert.isTrue(StringUtils.hasText(ids), "删除的数据为空");
        List<Long> idArray = Arrays.stream(ids.split(",")).map(Long::parseLong).collect(Collectors.toList());
        return this.removeByIds(idArray);
     }

    @Override
    public void importData(List<KafkaConsumeRecordImportVo> voList)  {
        List<KafkaConsumeRecord> kafkaConsumeRecordList = new ArrayList<>();
        for (KafkaConsumeRecordImportVo importVo : voList) {
            KafkaConsumeRecord kafkaConsumeRecord = new KafkaConsumeRecord();
            BeanUtils.copyProperties(importVo, kafkaConsumeRecord);
            kafkaConsumeRecordList.add(kafkaConsumeRecord);
        }
        this.saveOrUpdateBatch(kafkaConsumeRecordList);
    }

    @Override
    public List<KafkaConsumeRecordImportVo> findByIds(String idsStr) {
        if (!StringUtils.hasText(idsStr)) {
            return new ArrayList<>();
        }
        List<Long> ids = Arrays.stream(idsStr.split(",")).map(Long::parseLong).collect(Collectors.toList());
        List<KafkaConsumeRecord> kafkaConsumeRecordList = this.getBaseMapper().selectBatchIds(ids);
        List<KafkaConsumeRecordImportVo> voList = new ArrayList<>();
        for (KafkaConsumeRecord kafkaConsumeRecord : kafkaConsumeRecordList) {
            KafkaConsumeRecordImportVo vo = new KafkaConsumeRecordImportVo();
            BeanUtils.copyProperties(kafkaConsumeRecord, vo);
            voList.add(vo);
        }
        return voList;
    }
}
